package com.parking.bang.rpc.core.server;

import java.nio.channels.spi.SelectorProvider;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;

import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.parking.bang.rpc.core.thread.RPCThreadFactory;
import com.parking.bang.rpc.core.thread.RPCThreadPool;
import com.parking.bang.rpc.model.MessageHolder;
import com.parking.bang.rpc.model.ReqMessage;
import com.parking.bang.rpc.model.RespMessage;
import com.parking.bang.rpc.serialize.base.RPCSerializeProtocol;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

/**
 * Server Executor plus Spring
 * 
 * @author ivan
 *
 */
public class RPCServerExecutor implements ApplicationContextAware, InitializingBean {
	// as better to be Configuration with IOC
	private RPCSerializeProtocol rpcSerializeProtocol = RPCSerializeProtocol.JDK;
	// ip config
	private String serverAddress;
	private static final String DELIMITER = ":";
	// handler map
	private Map<String, Object> handlerMap = new ConcurrentHashMap<String, Object>();
	// executors
	private static ListeningExecutorService threadPoolExecutor;

	public RPCServerExecutor(RPCSerializeProtocol rpcSerializeProtocol, String serverAddress) {
		this.serverAddress = serverAddress;
		this.rpcSerializeProtocol = rpcSerializeProtocol;
	}

	public RPCServerExecutor(String rpcSerializeProtocol, String serverAddress) {
		this.serverAddress = serverAddress;
		this.rpcSerializeProtocol = Enum.valueOf(RPCSerializeProtocol.class, rpcSerializeProtocol);
	}

	/**
	 * run
	 */
	public void afterPropertiesSet() throws Exception {
		ThreadFactory threadRpcFactory = new RPCThreadFactory("NettyRPC ThreadFactory");

		int parallel = Runtime.getRuntime().availableProcessors() * 2;

		EventLoopGroup boss = new NioEventLoopGroup();
		EventLoopGroup worker = new NioEventLoopGroup(parallel, threadRpcFactory, SelectorProvider.provider());

		try {
			ServerBootstrap bootstrap = new ServerBootstrap();
			bootstrap.group(boss, worker).channel(NioServerSocketChannel.class)
					.childHandler(new RPCServerChannelInitializer(handlerMap).setProtocol(rpcSerializeProtocol))
					.option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true);

			String[] ipAddr = serverAddress.split(RPCServerExecutor.DELIMITER);

			if (ipAddr.length == 2) {
				String host = ipAddr[0];
				int port = Integer.parseInt(ipAddr[1]);
				ChannelFuture future = bootstrap.bind(host, port).sync();
				System.out.printf("Netty RPC Server start success!\nip:%s\nport:%d\nprotocol:%s\n\n", host, port,
						rpcSerializeProtocol);
				future.channel().closeFuture().sync();
			} else {
				System.out.printf("Netty RPC Server start fail!\n");
			}
		} finally {
			worker.shutdownGracefully();
			boss.shutdownGracefully();
		}
	}

	/**
	 * set handler Map
	 */
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		try {
			MessageHolder keyVal = (MessageHolder) applicationContext
					.getBean(Class.forName("com.parking.bang.rpc.model.MessageHolder"));
			Map<String, Object> rpcServiceObject = keyVal.getMessHold();

			Set<Entry<String, Object>> s = rpcServiceObject.entrySet();
			Iterator<Map.Entry<String, Object>> it = s.iterator();
			Map.Entry<String, Object> entry;

			while (it.hasNext()) {
				entry = it.next();
				handlerMap.put(entry.getKey(), entry.getValue());
			}
		} catch (ClassNotFoundException ex) {
			ex.printStackTrace();
		}

	}

	/**
	 * submit task
	 * 
	 * @param task
	 * @param ctx
	 * @param request
	 * @param response
	 */
	public static void submit(Callable<Boolean> task, ChannelHandlerContext ctx, ReqMessage request,
			RespMessage response) {
		if (threadPoolExecutor == null) {
			synchronized (RPCServerExecutor.class) {
				if (threadPoolExecutor == null) {
					threadPoolExecutor = MoreExecutors
							.listeningDecorator((ThreadPoolExecutor) RPCThreadPool.getExecutor(16, -1));
				}
			}
		}

		ListenableFuture<Boolean> listenableFuture = threadPoolExecutor.submit(task);

		Futures.addCallback(listenableFuture, new FutureCallback<Boolean>() {
			public void onSuccess(Boolean result) {
				ctx.writeAndFlush(response).addListener(new ChannelFutureListener() {
					public void operationComplete(ChannelFuture channelFuture) throws Exception {
						System.out.println("RPC Server Send message-id respone:" + request.getMessId());
					}
				});
			}

			public void onFailure(Throwable t) {
				t.printStackTrace();
			}
		}, threadPoolExecutor);
	}

}
